package com.andnnl.pool.user;

import com.andnnl.pool.Msg;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.function.ToIntFunction;

/**
 * 优先级用户消息，大量优先级的情况下用
 * Created by chenss on 2021/6/23.
 */
public class PriorityUser<T> extends AbsUser<T> {
    //优先级别
    private final PriorityBlockingQueue<Msg<T>> queue = new PriorityBlockingQueue<>(10
            , Comparator.comparingInt(Msg::getPriority));

    public PriorityUser(String name) {
        this.name = name;
    }
    public PriorityUser(String name,int batchSize) {
        this.name = name;
        this.batchSize = batchSize;
    }

    @Override
    public List<Msg<T>> takeMsg() {
        List<Msg<T>> list = new ArrayList<>();
        //忽略处理时，返回空
        if (isPause()) {
            return list;
        }
        int i = 0;
        while (i++ < batchSize) {
            Msg<T> m = queue.poll();//先从优先队列取
            if (m == null) {
                break;
            }
            dataSize.addAndGet(-m.getDataSize());//累计总的dataSize
            list.add(m);
        }
        return list;
    }

    @Override
    public boolean isAllEmpty() {
        return queue.isEmpty();
    }

    @Override
    public void addMsg(Msg<T> msg) {
        msg.setUser(this);
        dataSize.addAndGet(msg.getDataSize());//累计总的dataSize
        queue.add(msg);
    }





    public PriorityBlockingQueue<Msg<T>> getQueue() {
        return queue;
    }


    @Override
    public int getTotalSize() {
        return queue.size();
    }



}
